---
sidebar_position: 2.5
description: 基于SpringCLoudStream构建消息驱动组件
---
# 消息驱动组件
## 集成 Kafka 构建消息驱动微服务
### 安装 kafka
* 下载 kafka ： https://kafka.apache.org/quickstart
* 解压、启动 ZK 和 kafka Server 即可（使用默认配置）
![](./img/kafka工作流程图.png)
### 创建 SpringBoot 工程
创建SpringBoot工程，并引入依赖
```xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>pub.wkq</groupId>
    <artifactId>study</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>wkq-study-springboot</name>
    <description>wkq SpringBoot</description>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <!-- SpringBoot 监控端点 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- 让 SpringBoot 能够识别 bootstrap.yml -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-context</artifactId>
            <version>2.2.6.RELEASE</version>
        </dependency>
        <!-- Kafka -->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.5.0.RELEASE</version>
        </dependency>
        <!-- RocketMQ -->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.11</version>
        </dependency>
        <!-- MySQL 驱动 -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.19</version>
        </dependency>
        <!-- Spring Data Jpa -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-jpa</artifactId>
        </dependency>
    </dependencies>

    <build>
        <finalName>${artifactId}</finalName>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <goals>
                            <goal>repackage</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>
```
创建 application.yaml 和 bootstrap.yaml 文件并填充内容
```yaml title='bootstrap.yaml'
spring:
  profiles:
    # prod, dev
    active: dev
  application:
    name: study

# 暴露端点
management:
  endpoints:
    web:
      exposure:
        include: '*'
  endpoint:
    health:
      show-details: always

```
```yaml title='application.yaml'
server:
  port: 4000
  servlet:
    context-path: /study-dev

spring:
  # SpringBoot 集成 Kafka 的配置, 最低配置只需要配置 spring.kafka.bootstrap-servers
  kafka:
    bootstrap-servers: 60.205.142.63:9092
  jpa:
    show-sql: true
    hibernate:
      ddl-auto: none
    properties:
      hibernate.show_sql: true
      hibernate.format_sql: true
    open-in-view: false
  datasource:
    # 数据源
    url: jdbc:mysql://47.94.9.237:3306/wkq_study_springboot?autoReconnect=true&useUnicode=true&characterEncoding=utf8&useSSL=false
    username: root
    password: Qwe2022!
    type: com.zaxxer.hikari.HikariDataSource
    driver-class-name: com.mysql.cj.jdbc.Driver
    # 连接池
    hikari:
      maximum-pool-size: 8
      minimum-idle: 4
      idle-timeout: 30000
      connection-timeout: 30000
      max-lifetime: 45000
      auto-commit: true
      pool-name: WkqStudySpringBoot
#    consumer:
      # 如果 Consumer 没有指定 group-id, 则使用配置文件中配置的; 如果配置文件中也没有定义, 则由框架随机生成
#      group-id: imooc-study-ecommerce
#      auto-offset-reset: latest
#      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
#      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
#    producer:
#      key-serializer: org.apache.kafka.common.serialization.StringSerializer
#      value-serializer: org.apache.kafka.common.serialization.StringSerializer

# RocketMQ 的配置, 这是最低配置
#rocketmq:
#  name-server: 127.0.0.1:9876
#  producer:
    # 发送同一类消息的设置为同一个 group, 保证唯一
#    group: study
```
## 集成 RocketMQ 构建消息驱动微服务
* 下载 RocketMQ:https://rocketmq.apache.org/dowloading/releases/
* 下载以 bin-release 结尾的 zip 包解压即完成安装
## SpringCloud Stream消息驱动组件
### 为什么需要SpringCloud Stream
![](./img/为什么需要SpringCloudStream.png)
:::tip
如果代码里面可以直接与 kafka 、Rocket 实现交互，但是Kafka 与RocketMQ的代码是不同的，或者说任何消息中间件的代码都不可能相同 ，这无疑会带来额外的工作量，这是因为业务代码和消息中间件耦合在一起，于是SpringCloud Stream 出现了。
:::
引入 SpringCloudStream 之后，业务代码不直接用消息中间件交互，而是与SpringCloudStream实现交互。

官方定义SpringCloud Stream 是一个构建消息驱动微服务的框架，提供了一种解耦合的方式，让业务代码不直接接触消息中间件，而是使用SpringCloud Stream 的 API 去完成消息的生产和消费，
而真正与 kafka 、Rocket 等消息中间件交互的是Stream 的底层API

**SpringCloud Stream 实现了消息中间件和业务代码的解耦**，实现一套代码使用任意中间件（只需要做配置变更就可以）。
### SpringCloud Stream 应用模型
#### SpringCloud Stream 中的核心概念
![](./img/SpringCloudStream应用模型.png)
* 负责与中间件交互的抽象绑定器，可以认为是 kafka中的 topic
* 发送消息和接受消息的应用通信信道：Input、Output

#### SpringCloud Stream 发布——订阅模型
![](./img/SpringCloudStream发布-订阅模型.png)
* Topic 可以认为就是 Kafka 中的Topic 概念
* Producer 通过 Input 信道发布消息到 Topic 上
* Consumer 通过 Output 信道消费 Topic 上的消息

## SpringCloud Stream消费分组
![](./img/Stream消息分组.png)

* 应用的不同实例放在一个消费者组中，每一条消息只会被组内的一个实例消费
* 消费者组的思想是通过多实例扩展服务吞吐量，且不会造成消息的重复消费

:::tip
在现实的微服务架构中，每一个微服务为了实现高可用和负载均衡，都会部署多个实例。<br/>
但在很多情况下，生产者发送消息给某个具体的微服务（集群会出现重复消费，因为订阅了同样的topic）是希望被消费一次，为了解决这个问题，所以引入了消费组的概念。<br/>
如果在同一个 topic 上的应用需要启动多个实例，可以通过 stream 提供的属性为应用指定一个组名，这样应用的多个实例在接收到消息的时候只会有一个成员真正的去完成消息的消费并进行处理。<br/>
默认情况下，没有为应用指定消费者组的时候，SpringCloud Stream 会为其分配一个独立的匿名消费者。如果同一个 destination 下的同一个应用都没有指定消费者组，当消息发送之后，所有的应用都会对其进行消费，造成消息的重复消费。<br/>

:::

### 消费分组的配置
对于消费分组的配置，只需要指定消费方的 group（`spring.cloud.stream.bindings.input.group`） 属性即可

## SpringCloud Stream消费分区
![](./img/Stream消费分区.png)
通过引入消费分组已经能够保证在多实例的情况下每个消息只能被组内的一个实例消费。但是消费组无法控制消息具体被哪一个实例消费。即同一个条消息多次到达之后可能是由不同的实例进行消费的。

但是会有一些业务场景，需要对一些具有相同特征的消息每次都被相同的消费实例处理，比如监控服务需要汇总数据，消息生产者可以为消息增加一个 id ，使得拥有这些 id 的消息每次都能被发送到一个特定的实例上实现累计统计的效果，否则这些数据就会被分散到不同的节点导致监控结果不一致的情况。

分区概念的引入就是为了解决这样的问题，当生产者将消息发送给多个消费者实例的时候，保证共同特征的消息始终是由同一个消费者实例接受和处理。<br/>

SpringCloud Stream 为分区提供了抽象概念，用来在消息中间件的上层去实现分区处理，所以它对于消息中间件自身是否实现了消息分区并不关心。对不具备分区功能的消息中间件（比如 RabbitMQ）也增加了分区功能的扩展<br/>

完全屏蔽底层中间件的同时，还做了相应的特性支持。<br/>
### 消费分区的配置
对于消费分区的配置，需要对生产者和消费者同时进行配置
1. 开启 stream 分区支持 ，spring.cloud.stream.instanceCount: 实例个数 （需要对 分区个数进行取余计算）
2. 当前消费者的索引，从 0 开始    spring.cloud.stream.instanceIndex: 0
3. 配置 input 、output
```yaml
spring:
  cloud:
    stream:
      # SpringCloud Stream + Kafka
      kafka:
        binder: # 绑定器，绑定到具体的中间件上
          brokers: 60.205.142.63:9092
          auto-create-topics: true  # 如果设置为false, 就不会自动创建Topic, 你在使用之前需要手动创建好
      # SpringCloud Stream + RocketMQ
      #      rocketmq:
      #        binder:
      #          name-server: 127.0.0.1:9876
      # 开启 stream 分区支持
      instanceCount: 1  # 消费者的总数
      instanceIndex: 0  # 当前消费者的索引
      bindings:
        # 默认发送方
        output:      # 这里用 Stream 给我们提供的默认 output 信道
          destination: commerce-stream-client-default    # 消息发往的目的地, Kafka 中就是 Topic
          content-type: text/plain    # 消息发送的格式, 接收端不用指定格式, 但是发送端要
          # 消息分区
          producer:
#            partitionKeyExpression: payload.author  # 分区关键字, payload 指的是发送的对象, author 是对象中的属性,属性不存在会报错
            partitionCount: 1   # 分区大小，如果创建 了多个 分区，
            # 使用自定义的分区策略, 注释掉 partitionKeyExpression
            partitionKeyExtractorName: wkqPartitionKeyExtractorStrategy  # bean 的名称
            partitionSelectorName: wkqPartitionSelectorStrategy
        # 默认接收方
        input:      # 这里用 Stream 给我们提供的默认 input 信道
          destination: commerce-stream-client-default
          group: commerce-wkq-default
#          # 消费者开启分区支持
          consumer:
            partitioned: true

        # wkq 发送方
        wkqOutput:
          destination: commerce-stream-client-wkq
          content-type: text/plain
        # wkq 接收方
        wkqInput:
          destination: commerce-stream-client-wkq
          group: commerce-wkq-wkq

```
## 消息驱动组件总结
### Spring Messaging 模块
Spring Messaging 是 Spring Framework 中的一个模块，其作用就是统一消息的编程模型。

![](./img/Spring Messaging 模块.png)

* Payload 是想要传递的消息本身，由于Message 模块定义的泛型，所以可以是任意的类型。
* Header 是消息头部（Map 类型），也就是消息的源数据，可以给消息增加一些标识，用来增强消息的特性。
* Message 模块的通信模型 ，Message Channel 用于去接受消息， send(Message) 方法可以将消息发送到消息通道里面。

### SpringCloud Stream 的核心概念与思想
#### 通过 Binder 组件实现与外部消息系统通信，屏蔽了底层中间件的使用细节

![](./img/SpringCloudStream应用模型.png)

* 通过 Binder 组件实现与外部消息系统通信，屏蔽了底层中间件的使用细节
* 消息分类映射为通信信道，可以为不同类的消息自定义通信信道，采用发布、订阅模式，可以有效降低消息生产者和消费者的耦合。

#### 消费分组：应用多实例部署，提高服务吞吐量，且不重复消费消息
![](./img/消费分组.png)
**基于相同主题的消费者，可以轻松的实现扩展**

在创建SpringCloud Stream应用的时候建议最好为其指定一个消费者组，防止对消息的重复处理。

很多时候忘记指定消费者组的时候，很容易出现数据不一致的情况。
#### 消息分区：是为了确保具有共同特征标识的数据由同一个消费者实例进行处理
![](./img/Stream消费分区.png)
通过引入消费组的概念，我们已经能够在多实例的情况下保证每个消息只被组内的一个实例消费，但是消费组无法控制消息具体被哪一个实例消费，同一个消息，多次到达之后可能是由不同的实例进行消费的。<br/>
但是有一些业务场景，需要对一些具有相同特征的消息每次都由同一个消费实例处理，比如 监控服务，监控服务的数据往往是串行化的，而且是顺序化的，所以它需要保证被一个特定的服务实例消费才能够展现出现监控的效果，否则这些数据就会分散到不同的节点上导致监控结果不一致甚至是错误的情况。<br/>
而 分区概念的引入就是为了解决这样的问题。

当生产者将消息发送给多个消费者时，保证拥有共同特征的消息数据始终是同一个消费者实例接受和处理的。

消息的共同特征可以给 Stream 提供一个key ，或者是可以自定义一个 key 的提取器，再去根据这个 key 做一个 partition 方面的转发。

Stream 为分区提供了通用的抽象实现，用来在消息中间件的上层实现分区处理，所以说它对消息中间件自身是否实现了消息分区并不关心，为不具备分区功能的消息中间件增加了分区功能的扩展，这是在应用层面实现的。

分组和分区都可以采用简单的配置去完成，这也是使用 SpringCloud Stream 的优势。

